package com.littlerpc.registry.zookeeper;

import lombok.Getter;
import org.apache.zookeeper.*;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static com.littlerpc.registry.zookeeper.ZookeeperConfig.ROOT_NODE;

/**
 * Created by ChengPeng on 2017/7/2.
 * 用于服务自动发现
 */
public class ZookeeperDiscovery {

    private CountDownLatch latch = new CountDownLatch(1);
    @Getter
    private Set<String> providerListCache = new LinkedHashSet<>();
    @Getter
    private volatile boolean isConnected = false;

    private ZooKeeper zk = null;

    /**
     * 构造方法，初始化zookeeper链接
     */
    public ZookeeperDiscovery(ZookeeperConfig zkConf) {
        String connectString = zkConf.getConnectString();
        int timeout = zkConf.getTimeout();

        try {
            zk = new ZooKeeper(connectString, timeout, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getState() == Event.KeeperState.SyncConnected) {
                        latch.countDown();
                    }
                }
            });
            //等待链接完成
            latch.await(5, TimeUnit.SECONDS);
            isConnected = true;
            discoverProvider();

        } catch (Exception e) {
            //ignore
        }
    }

    private Watcher watche = new Watcher() {
        @Override
        public void process(WatchedEvent event) {
            //只要节点改变，重新获取节点信息
            if (event.getType() == Event.EventType.NodeChildrenChanged) {
                System.out.println(event.getPath());
                discoverProvider();
            }
        }
    };

    public void discoverProvider() {
        if(!isConnected){
            return ;
        }

        try {
            List<String> children = zk.getChildren(ROOT_NODE, watche);
            System.out.println("provider:**********************" + children);

            providerListCache = children.stream()
                    .map((path) -> ZookeeperUtils.getData(zk,ROOT_NODE,path))
                    .filter(ZookeeperDiscovery::isNotEmpty)
                    .collect(Collectors.toSet());

        } catch (KeeperException | InterruptedException e) {
            e.printStackTrace();
        }

    }

    private static boolean isNotEmpty(String str){
        return !str.isEmpty();
    }

}
